package cn.tianyigps.tianyigpskafka.task.customer;



import cn.tianyigps.tianyigpskafka.task.bean.Task;
import cn.tianyigps.tianyigpskafka.task.util.TaskUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.*;

/**
 * Created with IDEA
 *
 * @description: 消费者基类
 * @author: cui.guibin
 * @create: 2019/3/18 9:40
 **/
@Component
@Slf4j
public abstract class Customer {

    /**
     * @param: Task task
     * @return
     * @author cui.guibin
     * @date 2019/3/18 9:42
     */
     public abstract boolean checkTask(Task task);

     /**
      * @author cui.guibin
      * @date 2019/3/18 10:24
      * task 动态消费不同数据
      */
    @KafkaListener(topics ={"#{'${tianyi.kafka.topic}'.split(',')}"}, containerFactory = "batchContainerFactory")
    public void receive(List<ConsumerRecord<String, String>> records){
        for (ConsumerRecord<String, String> record : records) {
            String value = record.value();
            if (StringUtils.isEmpty(value)){
                log.error("kafka数据为空！");
                continue;
            }
            Task task = TaskUtil.json2Task(value);
            if(Optional.ofNullable(task).isPresent()){
                boolean b = this.checkTask(task);
                if (b)
                    this.consume(task);
                else {
                    log.error("数据参数不符： " + task );
                }
            }
        }
    }

    /**
     * @param: ConsumerRecord<String,String> record
     * @author cui.guibin
     * @date 2019/3/18 10:19
     * 消费到数据
     */ 
    public abstract void consume(Task task);

//    private Task getTask() {
//        //            Task task = TaskUtil.json2Task(value);
//        Map<String, Object> parentParams = new HashMap<>();
//        List<String> equipmentIds =  new ArrayList<>();
//        equipmentIds.add("14353564666");
//        equipmentIds.add("1435356645666");
//        parentParams.put("equipmentIds",equipmentIds);
//        Map<String, Object> sonParams = new HashMap<>();
//        sonParams.put("callback","http://test.mars.yikong.tianyigps.com/ts-driver-config/driver-config/13301607542");
//        return new Task("tasktest",new Task(CallbackTopicConstant.IMMEDIATE_TOPIC,null,sonParams),parentParams);
//    }

}
